package cn.doitedu.rtdw.data_sync;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job2_地理位置维表加工并加载到hbase {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 1. 建表，映射mysql中的地理位置维表： gps_tmp
        tenv.executeSql(
                " create table gps_tmp_mysql(                        "+
                        "   lat double,                                       "+
                        " 	lng double,                                      "+
                        " 	region string,                                   "+
                        " 	city string,                                     "+
                        " 	province string                                  "+
                        " ) with (                                           "+
                        "    'connector' = 'jdbc',                           "+
                        "    'url' = 'jdbc:mysql://doitedu:3306/realtimedw', "+
                        "    'table-name' = 'gps_tmp',                       "+
                        "    'username' = 'root',                            "+
                        "    'password' = 'root'                             "+
                        " )                                                  "
        );


        // 2. 建表，映射hbase中的目标表： dim_geo_area
        tenv.executeSql(
                " create table dim_geo_area_hbase(                    "+
                        "   geohash string,                                    "+
                        " 	f row<province string,city string,region string>,  "+
                        "   primary key(geohash) not enforced      "+
                        " ) with (                                 "+
                        "     'connector' = 'hbase-2.2',           "+
                        "     'table-name' = 'dim_geo_area',       "+
                        "     'zookeeper.quorum' = 'doitedu:2181'  "+
                        " )                                         "
        );

        // 3. 写一个insert .... select .. 的sql
        tenv.createTemporaryFunction("geocode", Gps2GeoHashUdf.class);
        tenv.executeSql(
                " INSERT INTO dim_geo_area_hbase      "+
                        " SELECT                              "+
                        "   geocode(lng,lat) as geohash,    "+
                        " 	row(province,city,region) as f    "+
                        " FROM gps_tmp_mysql                  "+
                        " WHERE  geocode(lng,lat) is not null "
        );



    }
}
